package top.atcra.rpc.proxy;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.lang.Pair;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import top.atcra.rpc.RPCBootstrap;
import top.atcra.rpc.annotation.RetryPolicy;
import top.atcra.rpc.compress.CompressorType;
import top.atcra.rpc.config.ConfigLoader;
import top.atcra.rpc.protection.circuitBreaker.CircuitBreaker;
import top.atcra.rpc.serialization.SerializerType;
import top.atcra.rpc.transport.message.RPCRequest;
import top.atcra.rpc.transport.message.RequestPayload;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程都封装在了invoke方法中
 * 1. 发现可用服务    2. 建立连接    3. 发送请求  4. 接收响应
 *
 * @author atcra
 * @date 2025-3-25
 **/
@Slf4j
public class RPCInvocationHandler implements InvocationHandler {

    private final String serviceName;
    private final String serviceGroup;
    private final RPCBootstrap bootstrap;
    private final Map<InetSocketAddress, Channel> channelCache;

    public RPCInvocationHandler(String serviceName, String serviceGroup, RPCBootstrap bootstrap) {
        this.serviceName = serviceName;
        this.serviceGroup = serviceGroup;
        this.bootstrap = bootstrap;
        this.channelCache = RPCBootstrap.CHANNEL_CACHE.get(bootstrap);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        // 1、利用方法、参数列表封装请求负载
        // 2、封装请求对象
        // 3、选取一个服务提供方
        // 4、建立连接
        // 5、写出请求
        // 6、等待响应（网络通信和方法调用是异步的）
        // 7、获得结果返回
        // 从接口中获取判断是否需要重试
        RetryPolicy tryTimesAnnotation = method.getAnnotation(RetryPolicy.class);

        // 默认尝试1次
        int tryTimes = 2;
        int intervalTime = 0;
        if (tryTimesAnnotation != null) {
            tryTimes = tryTimesAnnotation.retryTimes() + 1;
            intervalTime = tryTimesAnnotation.intervalTime();
        }
        for (int i = 0; i < tryTimes; ) {
            try {
                var result = invoke0(method, args);
                 System.out.println(result);
            } catch (Throwable e) {
                if (++i == tryTimes) {
                    log.error("对方法【{}】进行远程调用时，重试超过最大允许次数{}",
                            method.getName(), tryTimes, e);
                } else {
                    log.error("方法调用【{}】在进行第{}次尝试时发生异常.", method.getName(), i, e);
                }

                try {
                    Thread.sleep(intervalTime);
                } catch (InterruptedException ex) {
                    log.error("在进行重试时发生Interrupted异常.", ex);
                }
            }
        }

        throw new RuntimeException("执行远程方法" + method.getName() + "调用失败。");
    }
//        throw new UnsupportedOperationException("Not implement yet.");

    private Object invoke0(Method method, Object[] args) throws Throwable {
        var cfg = ConfigLoader.getConfig();
        RequestPayload requestPayload = RequestPayload.builder()
                .serviceName(serviceName)
                .group(serviceGroup)
                .methodName(method.getName())
                .parametersType(method.getParameterTypes())
                .parametersValue(args)
                .returnType(method.getReturnType())
                .build();
        RPCRequest request = RPCRequest.builder()
                .requestId(bootstrap.getRequestIdGenerator().getId2())
                .compressType(CompressorType.valueOf(cfg.getCompressor()).getId())
                .requestType(RPCRequest.RequestType.NORMAL.getId())
                .serializeType(SerializerType.valueOf(cfg.getSerializer()).getId())
                .timestamp(System.currentTimeMillis())
                .body(requestPayload)
                .build();
//        bootstrap.REQUEST_THREADLOCAL.set(request);
        // 1、选取一个服务提供方
        var address = bootstrap.getLoadBalancer().getAvailableService(serviceName, serviceGroup, request.getRequestId());
//        if (log.isDebugEnabled()) {
//            log.debug("服务调用方，发现了服务【{}】分组{}的可用主机【{}】.",
//                    serviceName,serviceGroup, address);
//        }
        // 服务熔断
        CircuitBreaker circuitBreaker = RPCBootstrap.HOST_CIRCUIT_BREAKERS.get(address);
        if (circuitBreaker == null) {
            circuitBreaker = ConfigLoader.getConfig().getCircuitBreaker();
            RPCBootstrap.HOST_CIRCUIT_BREAKERS.put(address, circuitBreaker);
        }

        // 如果断路器是打开的，则不发送请求（心跳包例外）
        if (request.getRequestType() != RPCRequest.RequestType.HEALTH_DETECT.getId()
            &&circuitBreaker.getState() == CircuitBreaker.State.OPEN) {
            circuitBreaker.delayReset(System.currentTimeMillis(), 1000); // 3s后重置为HALF_OPEN状态
            throw new RuntimeException("当前断路器已经开启，无法发送请求");
        }

        // 获取连接
        var channel = getAvailableChannel(address);

        if (log.isDebugEnabled()) {
            Assert.isTrue(channel.isActive(), "Channel.isActive=False。");
            Assert.isTrue(channel.isWritable(), "Channel.isWritable=False。");
            log.debug("成功获取服务{}({})的Channel,localAddr:{},remoteAddr:{}", serviceName, serviceGroup, channel.localAddress(), channel.remoteAddress());
        }

        CompletableFuture<Object> callFuture = new CompletableFuture<>();
        // 写出请求并等待响应
        var PENDING_CALLS = channel.attr(RPCBootstrap.PENDING_CALLS_KEY).get();
        PENDING_CALLS.put(request.getRequestId(), Pair.of(callFuture, request.getTimestamp()));

        channel.writeAndFlush(request).addListener((ChannelFutureListener) writtenPromise -> {
            // 注意当前的promise是writeAndFlush的返回结果，只需要处理以下异常就行了
            if (!writtenPromise.isSuccess()) {
                callFuture.completeExceptionally(writtenPromise.cause());
            }
        });
        // 断路器记录调用结果
        circuitBreaker.logCallResult(callFuture);
        try {
            return callFuture.get(cfg.getMaxWaitTimeMs(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            log.error("等待响应超时:{}", e.getMessage());
            throw e;
        }
        finally {
            // 移除PEDDING_CALLS中的请求
            PENDING_CALLS.remove(request.getRequestId());
        }
    }

    /**
     * 根据地址获取一个可用的通道
     *
     * @param address 服务地址
     * @return 可用通道
     */
    private Channel getAvailableChannel(InetSocketAddress address) {
        // 1、尝试从缓存中获取
        Channel channel = channelCache.get(address);

        // 2、拿不到就去建立连接
        if (channel != null && !(channel.isOpen() && channel.isWritable())) {
            channel.close();
            channel = null;
        }
        if (channel == null) {
            // 阻塞获取channel
            channel = bootstrap.createChannel(address);
            log.info("创建服务{}({})的Channel,localAddr:{},remoteAddr:{}", serviceName, serviceGroup, channel.localAddress(), channel.remoteAddress());

            // 缓存channel
            channelCache.put(address, channel);
        }

        return channel;
    }

}

